package cc.xpcas.raft.context;

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cc.xpcas.raft.component.SimpleThreadFactory;
import cc.xpcas.raft.message.MessageContext;

/**
 * @author xp
 */
public class MainlineExecutor {

    private static final Logger LOG = LoggerFactory.getLogger(MainlineExecutor.class);

    private static final AtomicInteger poolCounter = new AtomicInteger(0);

    private final Context context;

    private final ExecutorService executor;

    private AtomicInteger backlog = new AtomicInteger(0);

    private Thread thread;

    private MessageHandler messageHandler;

    public MainlineExecutor(Context context) {
        this.context = context;
        executor = Executors.newSingleThreadExecutor(new SimpleThreadFactory("raft-mainline-" + poolCounter.incrementAndGet()));
        Future future = executor.submit(() -> {
            // warm up
            this.thread = Thread.currentThread();
        });
        try {
            future.get();
        } catch (Exception e) {
            // should not happen
            throw new RuntimeException(e);
        }

        this.messageHandler = new MessageHandler(context);
    }

    public boolean isRunning() {
        return executor.isShutdown();
    }

    public boolean isRunningOnMainline() {
        return Thread.currentThread() == thread;
    }

    public void runOnMainline(Runnable origin) {
        Runnable runnable = () -> {
            backlog.decrementAndGet();
            Thread.currentThread().isInterrupted(); // clear interrupted flag
            origin.run();
            tryHandleQueuedMessagesOnMainline();
        };
        backlog.incrementAndGet();
        executor.execute(runnable);
    }

    public void enqueueMessage(MessageContext messageContext) {
        messageHandler.enqueue(messageContext);
        tryHandleQueuedMessagesOnMainline();
    }

    private void tryHandleQueuedMessagesOnMainline() {
        if (backlog.get() == 0) {
            executor.execute(() -> {
                try {
                    while (backlog.get() == 0) {
                        if (!messageHandler.runOnce()) break;
                    }
                } catch (InterruptedException e) {
                    // suppress
                }
            });
        }
    }

    public void interruptMainline() {
        if (isRunningOnMainline()) {
            throw new IllegalStateException("running on mainline");
        }
        thread.interrupt();
    }

    public void close() {
        executor.shutdown();
        try {
            while (executor.isTerminated()) {
                executor.awaitTermination(100, TimeUnit.MILLISECONDS);
            }
        } catch (InterruptedException e) {
            // suppress
        }
    }
}
